package com.parallel.compute;

import java.util.List;
import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
 * Read list data with different thread. The from index and to index of the list are calculated.
 * Put list data to queue.
 * Read queue with another thread.
 * @author issuser
 *
 */
public class QueueHandler {

	private List<Integer> data;
	private Queue<Integer> tempData;

	public QueueHandler() {
		initQueue(10);
		tempData = new ArrayBlockingQueue<Integer>(10);
	}

	public static void main(String[] strs) throws Exception {
		QueueHandler queueHandler = new QueueHandler();
		ExecutorService exec = Executors.newFixedThreadPool(10);
		exec.execute(queueHandler.new QueueRead(queueHandler.data,
				queueHandler.tempData, 2));
		for (int i = 0; i < 10; i++) {
			exec.execute(queueHandler.new QueueHandle(queueHandler.tempData));
		}
		TimeUnit.MILLISECONDS.sleep(2000);
		exec.shutdownNow();

	}

	private List<Integer> initQueue(final int size) {
		data = new CopyOnWriteArrayList<Integer>();
		for (int i = 0; i < size; i++) {
			data.add(i);
		}
		return data;
	}

	/**
	 * Put data to tempData.
	 * @author issuser
	 *
	 */
	class QueueRead implements Runnable {
		private List<Integer> data;
		private Queue<Integer> tempData;
		private int size;

		public QueueRead(List<Integer> data, Queue<Integer> tempData, int size) {
			this.data = data;
			this.tempData = tempData;
			this.size = size;
		}

		public void run() {
			try {
				while (true) {
					TimeUnit.MILLISECONDS.sleep(100);
					for (int i = 0; i < size; i++) {
						subThread(i);
					}
				}
			} catch (InterruptedException e) {
				//System.out.println("sleep() interrupted");
			}
		}

		private synchronized void subThread(int i) {
			if (data != null && data.size() > 0) {
				int start = data.size() / size * i;
				int end = data.size() / size * (i + 1);
				if (i == size - 1) {
					end = data.size();
				}
				final List<Integer> subData = data.subList(start, end);
				new Thread(new Runnable() {
					@Override
					public void run() {
						synchronized (this) {
							for (int j = 0; j < subData.size(); j++) {
								if(tempData.size() != 10) {
									tempData.add(subData.get(j));
								}
							}
						}
					}
				}).run();
				data.removeAll(subData);
			}
		}
	}

	/**
	 * Print tempData.
	 * @author issuser
	 *
	 */
	class QueueHandle implements Runnable {
		private Queue<Integer> tempData;

		public QueueHandle(Queue<Integer> tempData) {
			this.tempData = tempData;
		}

		public void run() {
			try {
				while (true) {
					TimeUnit.MILLISECONDS.sleep(200);
					if (!tempData.isEmpty()) {
						System.out.println(tempData.poll());
					}

				}
			} catch (InterruptedException e) {
				//System.out.println("sleep() interrupted");
			}
		}
	}
}
